package org.wejob.core.log;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class ProduceFuture implements Future<String> {

    MessageRecord record;
    Topic topic;
    int curMaxOffset=-1;
    String result;
    ReentrantLock futureLock=new ReentrantLock();
    Condition waiter;
    public ProduceFuture(MessageRecord record,Topic topic){
        this.record=record;
        this.topic=topic;
    }
    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        return false;
    }

    @Override
    public boolean isCancelled() {
        return false;
    }

    @Override
    public boolean isDone() {
        if(this.record.offset<0l)
            return false;
        return this.record.offset<=this.topic.maxCommitedOffset&&this.result!=null;
    }

    @Override
    public String get() throws InterruptedException, ExecutionException {
        this.futureLock.lock();
        if(!isDone()){
            this.waiter=this.futureLock.newCondition();
            System.out.println("producer thread--"+Thread.currentThread().getName()+" wait");
            this.waiter.await();
        }
        this.futureLock.unlock();
        return this.result;
    }

    @Override
    public String get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        //ToDo 这里使用信号量,当前线程应该阻塞到信号量上;等待结果完成--使用isDone()方法判断;
        Thread.currentThread().wait(timeout);
        return get();
    }
}
